package org.jetlinks.pro.device.message.writer;

import com.alibaba.fastjson.JSON;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import lombok.Generated;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.jetlinks.core.event.TopicPayload;
import org.jetlinks.core.message.DeviceMessage;
import org.jetlinks.pro.device.service.data.DeviceDataService;
import org.jetlinks.pro.gateway.DeviceMessageUtils;
import org.jetlinks.pro.gateway.annotation.Subscribe;
import org.jetlinks.pro.messaging.kafka.*;
import org.jetlinks.pro.utils.MessageTypeMatcher;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import reactor.core.Disposable;
import reactor.core.Disposables;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;
import reactor.util.retry.Retry;

import java.time.Duration;
import java.util.Collection;
import java.util.Collections;

@Slf4j

public class KafkaMessageWriterConnector implements CommandLineRunner {

    private final KafkaProducer producer;
    private final DeviceDataService dataService;

    private final KafkaProperties properties;

    private final Disposable.Composite disposable = Disposables.composite();

    @Getter
    @Setter
    private String topicName = "device.message";
    @Getter
    @Setter
    private boolean consumer = true;
    @Getter
    @Setter
    private MessageTypeMatcher type = new MessageTypeMatcher();
    @Getter
    @Setter
    private int buffer = 3000;
    @Getter
    @Setter
    private Duration bufferTimeout=Duration.ofSeconds(3);

    @Generated
    public KafkaMessageWriterConnector(DeviceDataService dataService,
                                       KafkaProperties properties) {
        this(dataService, new ReactorKafkaProducer(properties), properties);
    }

    public KafkaMessageWriterConnector(DeviceDataService dataService,
                                       KafkaProducer producer,
                                       KafkaProperties properties) {
        this.producer = producer;
        this.dataService = dataService;
        this.properties = properties;
    }

    public void shutdown() {
        disposable.dispose();
    }

    @Subscribe(topics = "/device/**", id = "device-message-kafka-writer")
    public Mono<Void> writeDeviceMessageToTs(TopicPayload payload) {

        ByteBuf topic = Unpooled.wrappedBuffer(payload.getTopic().getBytes());
        DeviceMessage message = payload.decode(DeviceMessage.class);
        ByteBuf messageBuf = Unpooled.wrappedBuffer(JSON.toJSONBytes(message.toJson()));
        if (!type.match(message.getMessageType())) {
            return Mono.empty();
        }

        return producer
            .send(Mono.just(SimpleMessage.of(topicName, topic, messageBuf)))
            .subscribeOn(Schedulers.boundedElastic());
    }

    private Mono<Void> doSaveDeviceData(Collection<DeviceMessage> messages) {
        return dataService
            .saveDeviceMessage(messages)
            .then(Mono.fromRunnable(() -> {
                log.debug("write device message from kafka [{}] success", messages.size());
            }))
            .retryWhen(Retry.fixedDelay(3, Duration.ofSeconds(2)))
            .onErrorResume((err) -> {
                log.error("write device data from kafka error", err);
                return Mono.empty();
            })
            .then();
    }

    @Generated
    protected KafkaConsumer createConsumer() {
        return new ReactorKafkaConsumer(Collections.singleton(topicName), properties);
    }

    @Override
    public void run(String... args) {
        if (!consumer) {
            return;
        }
        KafkaConsumer consumer = createConsumer();
        disposable
            .add(consumer
                     .subscribe()
                     .flatMap(msg -> Mono.justOrEmpty(DeviceMessageUtils.convert(msg.getPayload())))
                     .bufferTimeout(getBuffer(), getBufferTimeout())
                     .flatMap(this::doSaveDeviceData, Integer.MAX_VALUE)
                     .subscribe()
            );
    }
}
